from fastapi import APIRouter, Body, Request, WebSocket
from fastapi.responses import StreamingResponse
from typing import Dict, Any, Optional
import asyncio
from intelli_port.data_layer.clients import mysql_execute_read, mysql_execute_write
from intelli_port.commons.audit import audit_event
from intelli_port.notifications.websocket_hub import hub
from intelli_port.notifications.notifier import register_webhook, register_robot_webhook


router = APIRouter()


@router.post("/click")
async def click(payload: Dict[str, Any] = Body(...)):
    flight_no = str(payload.get("flight_no") or "").strip()
    detail = str(payload.get("detail") or "").strip()
    if not flight_no:
        return {"ok": False}
    try:
        r = mysql_execute_read(
            "SELECT id FROM flights WHERE flight_no=%s",
            (flight_no,),
            fetch="one",
        ).get("rows", [])
        if not r or not r[0] or r[0][0] is None:
            return {"ok": False}
        fid = int(r[0][0])
        mysql_execute_write(
            "INSERT INTO flight_events(flight_id, event, detail, created_at) VALUES(%s,'click',%s,NOW())",
            (fid, detail),
        )
        try:
            audit_event("user_event", "flight", "click", {"flight_no": flight_no})
        except Exception:
            pass
        return {"ok": True}
    except Exception:
        return {"ok": False}


@router.get("/flight/changes")
async def flight_changes(flight_no: str):
    try:
        r = mysql_execute_read("SELECT id FROM flights WHERE flight_no=%s", (flight_no,), fetch="one").get("rows", [])
        if not r or not r[0] or r[0][0] is None:
            return {"changes": []}
        fid = int(r[0][0])
        rows = mysql_execute_read(
            "SELECT event, detail, created_at FROM flight_events WHERE flight_id=%s ORDER BY created_at DESC LIMIT 50",
            (fid,),
            fetch="all",
        ).get("rows", [])
        changes = [{"event": x[0], "detail": x[1], "at": x[2].isoformat() if x[2] else None} for x in rows]
        return {"changes": changes}
    except Exception:
        return {"changes": []}


@router.get("/flight/stream")
async def flight_stream(request: Request, flight_no: Optional[str] = None, poll_interval_s: float = 1.0):
    async def _gen():
        last_ts: Optional[str] = None
        while True:
            if await request.is_disconnected():
                break
            try:
                if flight_no:
                    r = mysql_execute_read("SELECT id FROM flights WHERE flight_no=%s", (flight_no,), fetch="one").get("rows", [])
                    if not r or not r[0] or r[0][0] is None:
                        yield "data: {\"changes\": []}\n\n"
                        await asyncio.sleep(poll_interval_s)
                        continue
                    fid = int(r[0][0])
                    if last_ts:
                        rows = mysql_execute_read(
                            "SELECT event, detail, created_at FROM flight_events WHERE flight_id=%s AND created_at>%s ORDER BY created_at ASC",
                            (fid, last_ts),
                            fetch="all",
                        ).get("rows", [])
                    else:
                        rows = mysql_execute_read(
                            "SELECT event, detail, created_at FROM flight_events WHERE flight_id=%s ORDER BY created_at DESC LIMIT 10",
                            (fid,),
                            fetch="all",
                        ).get("rows", [])
                else:
                    if last_ts:
                        rows = mysql_execute_read(
                            "SELECT flight_id, event, detail, created_at FROM flight_events WHERE created_at>%s ORDER BY created_at ASC",
                            (last_ts,),
                            fetch="all",
                        ).get("rows", [])
                    else:
                        rows = mysql_execute_read(
                            "SELECT flight_id, event, detail, created_at FROM flight_events ORDER BY created_at DESC LIMIT 10",
                            fetch="all",
                        ).get("rows", [])
                changes = []
                if rows:
                    for x in rows:
                        if flight_no:
                            ev, dt, at = x[0], x[1], x[2]
                            changes.append({"event": ev, "detail": dt, "at": at.isoformat() if at else None})
                            last_ts = at.strftime("%Y-%m-%d %H:%M:%S") if at else last_ts
                        else:
                            fid, ev, dt, at = x[0], x[1], x[2], x[3]
                            changes.append({"flight_id": fid, "event": ev, "detail": dt, "at": at.isoformat() if at else None})
                            last_ts = at.strftime("%Y-%m-%d %H:%M:%S") if at else last_ts
                payload = {"changes": changes}
                import json as _json
                yield f"data: {_json.dumps(payload, ensure_ascii=False)}\n\n"
            except Exception:
                yield "data: {\"changes\": []}\n\n"
            await asyncio.sleep(max(0.2, float(poll_interval_s)))
    return StreamingResponse(_gen(), media_type="text/event-stream")


@router.websocket("/ws/flight/changes")
async def ws_flight_changes(ws: WebSocket):
    await ws.accept()
    try:
        q = ws.query_params
        flight_no = q.get("flight_no") or ""
        if not flight_no:
            await ws.close()
            return
        await hub.subscribe(flight_no, ws)
        while True:
            try:
                await ws.receive_text()
            except Exception:
                break
    finally:
        try:
            q = ws.query_params
            flight_no = q.get("flight_no") or ""
            if flight_no:
                hub.unsubscribe(flight_no, ws)
        except Exception:
            pass


@router.post("/webhook/register")
async def webhook_register(payload: Dict[str, Any] = Body(...)):
    flight_no = str(payload.get("flight_no") or "").strip()
    url = str(payload.get("url") or "").strip()
    secret = str(payload.get("secret") or "").strip()
    if not flight_no or not url:
        return {"ok": False}
    return register_webhook(flight_no, url, secret)


@router.websocket("/ws/robot/events")
async def ws_robot_events(ws: WebSocket):
    await ws.accept()
    try:
        q = ws.query_params
        robot_id = q.get("robot_id") or ""
        if not robot_id:
            await ws.close()
            return
        topic = f"robot:{robot_id}"
        await hub.subscribe(topic, ws)
        while True:
            try:
                await ws.receive_text()
            except Exception:
                break
    finally:
        try:
            q = ws.query_params
            robot_id = q.get("robot_id") or ""
            if robot_id:
                topic = f"robot:{robot_id}"
                hub.unsubscribe(topic, ws)
        except Exception:
            pass


@router.post("/webhook/register/robot")
async def webhook_register_robot(payload: Dict[str, Any] = Body(...)):
    rid = int(payload.get("robot_id") or 0)
    url = str(payload.get("url") or "").strip()
    secret = str(payload.get("secret") or "").strip()
    if rid <= 0 or not url:
        return {"ok": False}
    return register_robot_webhook(rid, url, secret)